package tableAndSql;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/8/6
 * @Update 2021/8/6 15:48
 */
public class KafkaToHbase {
    public static void main(String[] args) {

        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);

        bsTableEnv.executeSql("create table kafkaTable (`vehicle_color` STRING, `pass_time` BIGINT) " +
                "with ('connector' = 'kafka','topic' = 'ods.afterVehicle.vehicle.info','properties.bootstrap.servers' = " +
                "'172.17.1.101:9092','properties.group.id' = 'test4Group','scan.startup.mode' = 'earliest-offset','format' = 'json')");

        bsTableEnv.executeSql("create table hbaseTable(" +
                "pass_time BIGINT, test Row<vehicle_color STRING, pass_time BIGINT>,PRIMARY KEY (pass_time) NOT ENFORCED) " +
                "with ('connector' = 'hbase-1.4','table-name' = 'flink_test_demo'," +
                "'zookeeper.quorum' = '172.17.1.103:2181,172.17.1.102:2181,172.17.1.101:2181'," +
                "'zookeeper.znode.parent' = '/hbase-unsecure')");


        Table table2 = bsTableEnv.sqlQuery("select pass_time, Row(vehicle_color, pass_time) from kafkaTable");
        TableResult tableResult = table2.executeInsert("hbaseTable");
        tableResult.print();
    }
}
